package com.atguigu.chapter06;

import com.atguigu.chapter02.Event;
import com.atguigu.chapter05.Clicksource;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class CustomWatermaekTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.addSource(new Clicksource()).assignTimestampsAndWatermarks(new CustomWatermakrStrategy()).print();

        env.execute();
    }


    //自定义生成水位线
    private static class CustomWatermakrStrategy implements WatermarkStrategy<Event> {
        //告诉程序数据源里的时间戳是哪一个字段
        @Override
        public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new SerializableTimestampAssigner<Event>() {
                @Override
                public long extractTimestamp(Event event, long l) {
                    return event.timestamp;
                }
            };
        }

        @Override
        public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            return new CustomPeriodicGenerator();      //要什么给什么
        }
    }

    //周期性水位线生成器
    private static class CustomPeriodicGenerator implements WatermarkGenerator<Event> {
        private long delaytime = 5000L;
        private long maxTs = Long.MIN_VALUE+delaytime+1L;
        @Override
        public void onEvent(Event event, long l, WatermarkOutput watermarkOutput) {
            maxTs = Math.max(event.timestamp,maxTs);
        }

        @Override
        public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
            watermarkOutput.emitWatermark(new Watermark(maxTs - delaytime - 1L));
        }
    }

    //断点式水位线生成器
    public static class CustomPunctuatedGenerator implements WatermarkGenerator<Event>{

        @Override
        public void onEvent(Event event, long l, WatermarkOutput watermarkOutput) {
            if (event.user.equals("Mary")){
                watermarkOutput.emitWatermark(new Watermark(event.timestamp - 1));
            }
        }

        @Override
        public void onPeriodicEmit(WatermarkOutput watermarkOutput) {

        }
    }

}
